package zisu.netty.practice.RPC.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Desc: ""
 * @Author: caixiang
 * @DATE: 2021/1/9 17:19
 */
public class RpcConnectManager {

    //在多线程中 rpcConnectManager 这个共享资源内容很容易发生改变，所以要加volatile 修饰。
    private static volatile RpcConnectManager RPC_CONNECT_MANAGER = new RpcConnectManager();

    private RpcConnectManager(){

    }

    /**
     * 一个连接地址，对应一个实际的业务处理器（client）
     * */
    private Map<InetSocketAddress,RpcClientHandler> connectedHandlerMap = new ConcurrentHashMap<>();

    // 用于异步提交连接请求的 线程池。
    private ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(16,16 ,60 , TimeUnit.SECONDS, new ArrayBlockingQueue<>(1024));

    //所有连接成功的地址 所对应的 任务执行器列表。
    private CopyOnWriteArrayList<RpcClientHandler> connectedHandlerList = new CopyOnWriteArrayList<>();

    //
    private ReentrantLock connectedLock = new ReentrantLock();

    private Condition connectedCondition = connectedLock.newCondition();

    private EventLoopGroup eventLoopGroup = new NioEventLoopGroup(4);

    //最大连接时长，，超过这个时间 我就放弃了。
    private long connectTimeoutMills = 6000;

    private volatile AtomicInteger handlerIdx = new AtomicInteger(0);

    private volatile boolean isRunning = true;

    public static RpcConnectManager getInstance(){
        return RPC_CONNECT_MANAGER;
    }

    /**
     * @param serverAddress
     */
    //1. 异步连接线程池 真正的发起连接 ； 连接失败监听，连接成功监听。
    //2. 对于连接成功的资源做一个缓存（做一个管理）
    public void connect(final String serverAddress){
        List<String> allServerAddresses = Arrays.asList(serverAddress.split(","));
        updateConnectedServer(allServerAddresses);
    }

    /**
     * 更新缓存信息，并异步发起连接
     * @param allServerAddresses
     * 192.168.0.1:8765;;192.168.0.2:8765
     */
    public void updateConnectedServer(List<String> allServerAddresses){
        if(!allServerAddresses.isEmpty()){
            //1.解析 allServerAddresses，并临时存储到我们的newAll（HashSet集合中） 里面做缓存。
            //HashSet去重。
            HashSet<InetSocketAddress> newAll = new HashSet<InetSocketAddress>();
            for(int i=0;i<allServerAddresses.size();i++){
                String[] array = allServerAddresses.get(i).split(":");
                if(array.length == 2){
                    String host = array[0];
                    Integer port = Integer.parseInt(array[1]);
                    final InetSocketAddress remotePeer = new InetSocketAddress(host, port);
                    newAll.add(remotePeer);
                }
            }

            //2. 调用建立连接的方法，发起远程操作
            for(InetSocketAddress inetSocketAddress:newAll){
                if(!connectedHandlerMap.containsKey(inetSocketAddress)){
                    //没有建立连接 的才需要建立连接。
                    connectAsync(inetSocketAddress);
                }
            }

            //3.如果 allServerAddresses列表里不存在的连接地址，那么我们需要从缓存中进行移除。
            // （意思就是allServerAddresses 有A B C , 但是newAll缓存中 有 A B C D ,多了的这个D 我们要把它删除。）
            for(int i=0; i<connectedHandlerList.size(); i++){
                RpcClientHandler rpcClientHandler = connectedHandlerList.get(i);
                SocketAddress remotePeer = rpcClientHandler.getRemotePeer();
                //contains()  是 equal() ,而不是 “contains”
                if(!newAll.contains(remotePeer)){
                    System.out.println("remove invaild server node!");
                    RpcClientHandler handler = connectedHandlerMap.get(remotePeer);
                    if(handler!=null){
                        handler.close();
                        connectedHandlerMap.remove(remotePeer);
                    }
                    connectedHandlerList.remove(rpcClientHandler);
                }
            }
        }else {
            System.out.println("no available server address! ");
            //清除所有连接。
            clearConnected();
        }
    }


    /**
     * @param //inetSocketAddress 异步发起连接方法
     */
    private void connectAsync(InetSocketAddress remotePeer) {
        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                Bootstrap bootstrap = new Bootstrap();
                //group(eventLoopGroup)就是把当前的 连接 注册到 eventLoopGroup 事件循环组。
                bootstrap.group(eventLoopGroup)
                        .channel(NioSocketChannel.class)
                        .option(ChannelOption.TCP_NODELAY,true)  //禁用 nagle算法。数据包不做合并操作
                        //Nagle算法主要是避免发送小的数据包，要求TCP连接上最多只能有一个未被确认的小分组，在该分组的确认到达之前不能发送其他的小分组。
                        .handler(new RpcClientInitializer());
                connect(bootstrap,remotePeer);
            }
        });
    }

    private void connect(final Bootstrap b,InetSocketAddress remotePeer){
        // 1.真正建立连接；
        final ChannelFuture channelFuture = b.connect(remotePeer);

        // 2.连接失败的时候建立监听， 清除资源后进行发起重连操作；( closeFuture()当channel被关闭的时候 会返回ChannelFuture，，就是后面只监听
        // channel 关闭的operationComplete() 事件 )
        channelFuture.channel().closeFuture().addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                System.out.println("channelFuture.channel close operationComplete,remotePeer ="+remotePeer);
                //future下的eventLoop 就是 这里好连接的事件组
                future.channel().eventLoop().schedule(new Runnable() {
                    @Override
                    public void run() {
                        System.err.println("connect fail ,to reconnect!");
                        clearConnected();
                        //重连(回调方法)
                        connect(b,remotePeer);
                    }
                }, 3, TimeUnit.SECONDS);
            }
        });


        // 3.连接成功的时候添加监听， 把我们的新连接接入缓存
        channelFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if(future.isSuccess()){
                    System.out.println("successfully connect to remote server ,remote peer+"+remotePeer);
                    //还可以通过future 获取RpcClientHandler.class
                    RpcClientHandler rpcClientHandler = future.channel().pipeline().get(RpcClientHandler.class);
                    addHandler(rpcClientHandler);
                }
            }
        });
    }


    /**
     * @param rpcClientHandler 添加RpcClientHandler 到指定缓存
     * connectedHandlerMap & connectedHandlerList
     */
    public void addHandler(RpcClientHandler rpcClientHandler){
        connectedHandlerList.add(rpcClientHandler);
        //这些写 是有些问题的，rpcClientHandler.getRemotePeer() 虽然 active 了 但是 通道还没有激活，所以 获取远程地址 可能为空指针
        //InetSocketAddress remotePeer = (InetSocketAddress) rpcClientHandler.getRemotePeer();
        InetSocketAddress remotePeer = (InetSocketAddress) rpcClientHandler.getChannel().remoteAddress();
        connectedHandlerMap.put(remotePeer,rpcClientHandler);
        //signalAvailableHandler 唤醒可用的业务执行器
        signalAvailableHandler();
    }

    /**
     *  唤醒另外一端的线程（这段的线程一定是阻塞的）， 告知有新连接接入
     * */
    private void signalAvailableHandler() {
        connectedLock.lock();
        try {
            //唤醒所有
            connectedCondition.signalAll();
        }finally {
            connectedLock.unlock();
        }
    }

    /**
     * 连接失败时，及时释放资源，清空缓存
     * 先删除 所有的connectHandlerMap中的数据
     * 然后清空connectHandlerList 中的数据
     * */
    private void clearConnected(){
        for(RpcClientHandler rpcClientHandler : connectedHandlerList){
            //通过RpcClientHandler 找到具体的remotePeer，从connectedHandlerMap 进行移除指定的RpcClientHandler
            SocketAddress remotePeer = rpcClientHandler.getRemotePeer();
            RpcClientHandler handler = connectedHandlerMap.get(remotePeer);
            //只要把连接成功的连接 从 connectedHandlerMap 剔除就行了，其他不用剔除。
            if(handler!=null){
                handler.close();
                connectedHandlerMap.remove(remotePeer);
            }
        }
        connectedHandlerList.clear();
    }


    /**
     * @return 等待新连接接入 通知方法 (当前线程 会阻塞住，别的线程如果要调用 也只能等待)
     * @throws InterruptedException
     */
    private boolean waitingForAvailableHandler() throws InterruptedException {
        connectedLock.lock();
        try {
            return connectedCondition.await(this.connectTimeoutMills,TimeUnit.MICROSECONDS);
        }finally {
            connectedLock.unlock();
        }
    }

    /**
     * @return 选择一个实际的业务处理器
     */
    public RpcClientHandler chooseHandler(){
        //拿备份来操作，，不拿实际的数据来操作（复制出来的没有线程安全问题）(copy的并不是 同一份应用，而是不同的地址空间)
        CopyOnWriteArrayList<RpcClientHandler> handlers = (CopyOnWriteArrayList<RpcClientHandler>)this.connectedHandlerList.clone();
        Integer size = handlers.size();
        //当还在运行 的时候，size<=0 时候说明 没有已成功连接
        while (isRunning && size <= 0){
            try {
                boolean b = waitingForAvailableHandler();
                if(b){
                    //(copy的并不是 同一份应用，而是不同的地址空间),重新拷贝一下
                    handlers = (CopyOnWriteArrayList<RpcClientHandler>)this.connectedHandlerList.clone();
                    size  = handlers.size();
                }
            }catch (Exception e){
                System.err.println("wating for available node is interrupted !");
                throw new RuntimeException("no connected any server:"+e);
            }
        }
        if(!isRunning){
            return null;
        }

        //并没有 指定哪个索引，反正就是取余（ 主要作用：在一个集合中 随机选一个就行了。 ）
        int index = (handlerIdx.getAndAdd(1)+size)%size;
        return handlers.get(index);
    }


    public void stop(){
        isRunning = false;
        for (int i=0;i<connectedHandlerList.size();i++){
            RpcClientHandler handler = connectedHandlerList.get(i);
            handler.close();
        }
        //这里调用一下唤醒 操作
        signalAvailableHandler();
        threadPoolExecutor.shutdown();
        eventLoopGroup.shutdownGracefully();
    }


    /**
     * 发起重连的方法 要把对应的资源进行释放，然后重新连接
     * @param handler
     * @param remotePeer
     */
    public void reconnected(final RpcClientHandler handler,final SocketAddress remotePeer){
        if(handler != null){
            handler.close();
            connectedHandlerList.remove(handler);
            connectedHandlerMap.remove(remotePeer);
        }
        connectAsync((InetSocketAddress) remotePeer);

    }

}